package cn.doitedu.day08

import java.text.SimpleDateFormat

import scala.io.Source

/**
 * 流量统计，将两次流量请求时间小于10分钟的合并
 */
object Test03 {

  def main(args: Array[String]): Unit = {

    val source = Source.fromFile("data/test3.txt")

    val lines: Iterator[String] = source.getLines()

    val sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")

    val tpIt: Iterator[(String, Long, Long, Double)] = lines.map(line => {
      val fields = line.split(",")
      val uid = fields(0)
      val startTime = sdf.parse(fields(1)).getTime
      val endTime = sdf.parse(fields(2)).getTime

      val flow = fields(3).toDouble
      (uid, startTime, endTime, flow)
    })
    //分组(按照用户ID进行分组)
    val grouped: Map[String, List[(String, Long, Long, Double)]] = tpIt.toList.groupBy(_._1)
    val res: Iterable[(String, Long, Long, Double, Int)] = grouped.values.flatMap(lst => {

      //定义一个临时变量，保存上一条的endTime
      var temp = 0L
      var flag = 0
      var sum_flag = 0
      val sorted = lst.sortBy(_._2)
      sorted.map(t => {
        val uid = t._1
        val startTime = t._2
        val endTime = t._3
        val flow = t._4
        //在temp != 0 的情况，即不是sorted集合中的第一条
        if(temp != 0) {
          //判断差值是否大于10分钟
          if((startTime - temp) / (1000 * 60) > 10) {
            flag = 1
          } else {
            flag = 0
          }
        }
        //结束时间
        temp = t._3
        sum_flag += flag
        (uid, startTime, endTime, flow, sum_flag)
      })
    })

    //再分组，将用户ID和sum_flag相同的分到一组
    val grouped2: Map[(String, Int), Iterable[(String, Long, Long, Double, Int)]] = res.groupBy(t => (t._1, t._5))

//    val res2: Map[(String, Int), (Long, Long, Double)] = grouped2.mapValues(it => {
//      it.foldLeft((0L, 0L, 0.0))((a, b) => {
//        (Math.min(a._1, b._2), Math.max(a._2, b._3), a._3 + b._4)
//      })
//    })

    val res3 = grouped2.mapValues(it => {
      it.map(t => {
        (t._2, t._3, t._4) //起始时间，结束时间，流量
      }).reduce((a, b) => {
        (Math.min(a._1, b._1), Math.max(a._2, b._2), a._3 + b._3)
      })

    })


    println(res3)


    source.close()

  }

}
